package com.aotain.metis.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;

/**
 * 检测IP作为源的统计
 * @ClassName: AbnStatisBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年4月28日 下午1:33:32 
 *
 */
public class MonAbnReverseStatisBolt extends AbstractBolt{

	private static final long serialVersionUID = 1L;

	
	String dip = null;
	String dport = null;
	String sip = null;
	String sport = null;
	String dateStr_d = null;
	String dateStr_h = null;
	String dateStr_hour = null;
	String sourceAreaId = null;
	String areaName = null;
	String sourceAreaName = null;
	String idcno = null;
	String sourceCountry = null;
	Long triplePV = null;
	String abnormaltype = null;
	String pvKey = null;
	String sessKey = null;
	String sessReqKey = null;
	Long upStreamOctets = 0l;
	Long upStreamPacket = 0l;
	Long dnStreamOctets = 0l;
	Long dnStreamPacket = 0l; 

	private Map<String,Integer> pvMap = new HashMap<String,Integer>();
	private Map<String,Long> upStreamOctetsMap = new HashMap<String,Long>();//
	private Map<String,Long> upStreamPacketMap = new HashMap<String,Long>();//
	private Map<String,Long> dnStreamOctetsMap = new HashMap<String,Long>();//
	private Map<String,Long> dnStreamPacketMap = new HashMap<String,Long>();//
	private Map<String,Long> sessReqOctetsMap = new HashMap<String,Long>();
	

	

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(isTickTuple(tuple)) {
				emitCountingData(collector);  	
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println(" MonAbnReverseStatisBolt exception"+e.getMessage());
			Logger.getRootLogger().error("MonAbnReverseStatisBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}

	private void emitCountingData(OutputCollector collector) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);   


//		SimpleDateFormat sdfm = new SimpleDateFormat("mm");
//		SimpleDateFormat sdfh = new SimpleDateFormat("yyyyMMddHH00");
		SimpleDateFormat sdfh_ = new SimpleDateFormat("yyyyMMddHH");
//		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
//		String dateStrM = sdfm.format(date);
//		String dateStrH = sdfh.format(date);
		String dateStrH_ = sdfh_.format(date);
//		String dateStrD = sdfd.format(date);
//		int nMin5 = (Integer.parseInt(dateStrM)/5)*5;
//		int nMin30 = (Integer.parseInt(dateStrM)/30)*30;
//		String sMin5 = String.format("%02d", nMin5);
//		String sMin30 = String.format("%02d", nMin30);
//		String dateStr5 = dateStrH_+sMin5;
//		String dateStr30 = dateStrH_+sMin30;
//		String rowKeyAbnSession5Min = null;
//		String rowKeyAbnSession30Min = null;
//		String rowKeyAbnSessionH = null;
//		String rowKeyAbnSessionD = null;
//		String info = String.format("MetisStat<<<<<<<<<MapSize[%d]", upStreamOctetsMap.size());
//		Logger.getRootLogger().info(info);
		
		for(Entry<String,Long> entry : upStreamOctetsMap.entrySet()) {
			String entryKey = entry.getKey();
			String dateD = entryKey.split("_",-1)[0];
			String sip = entryKey.split("_",-1)[1];
			String type = entryKey.split("_",-1)[2];
			String key = dateD + "_" + sip;
			String hourKey = dateStrH_ + "_" + sip;
			
			/**
			 * 天表
			 */
			hbaseInstance.Add("METIS_SESS_STAT_DAY", key, "cf", "REPORTTIME", dateD);
			hbaseInstance.Add("METIS_SESS_STAT_DAY", key, "cf", "EVALUATE", "40");
			hbaseInstance.Add("METIS_SESS_STAT_DAY", key, "cf", "IP", sip);
			hbaseInstance.incerment("METIS_SESS_STAT_DAY", key, "cf:"+type, upStreamOctetsMap.get(entryKey));
			
			/**
			 * 小时表
			 */
			hbaseInstance.Add("METIS_SESS_STAT_HOUR", hourKey, "cf", "REPORTTIME", dateStrH_);
			hbaseInstance.Add("METIS_SESS_STAT_HOUR", hourKey, "cf", "EVALUATE", "40");
			hbaseInstance.Add("METIS_SESS_STAT_HOUR", hourKey, "cf", "IP", sip);
			hbaseInstance.incerment("METIS_SESS_STAT_HOUR", hourKey, "cf:"+type, upStreamOctetsMap.get(entryKey));
			
		}
		
		for(Entry<String,Long> entry:sessReqOctetsMap.entrySet()) {
			String entryKey = entry.getKey();
			long value = entry.getValue();
			String sip = entryKey.split("_",-1)[0];
			String dateD = entryKey.split("_",-1)[1];
			String dip = entryKey.split("_",-1)[2];
			String type = entryKey.split("_",-1)[3];
			String key = entryKey.substring(0,entryKey.lastIndexOf("_"));
			
			hbaseInstance.Add("METIS_SESS_REQ_STAT_DAY", key, "cf", "REPORTTIME", dateD);
			hbaseInstance.Add("METIS_SESS_REQ_STAT_DAY", key, "cf", "EVALUATE", "40");
			hbaseInstance.Add("METIS_SESS_REQ_STAT_DAY", key, "cf", "IP", sip);
			hbaseInstance.Add("METIS_SESS_REQ_STAT_DAY", key, "cf", "DESTIP", dip);
			hbaseInstance.incerment("METIS_SESS_REQ_STAT_DAY", key, "cf:"+type, value);
		}
		
		
//		pvMap.clear();
		upStreamOctetsMap.clear();
		sessReqOctetsMap.clear();
//		upStreamPacketMap.clear();
//		dnStreamOctetsMap.clear();
//		dnStreamPacketMap.clear();
	}

	private void countInLocal(Tuple tuple) {

		if(tuple != null ) {  
			dip = tuple.getStringByField("dip");
			sip = tuple.getStringByField("sip");
			idcno = tuple.getStringByField("idcno");
			dateStr_d = tuple.getStringByField("dateStr_d"); 
			triplePV = tuple.getLongByField("pv");
			abnormaltype = tuple.getStringByField("abntype");
			upStreamOctets = tuple.getLongByField("streamoctets");
			upStreamPacket = tuple.getLongByField("streampacket");
//			dateStr_h = tuple.getStringByField("dateStr_h");
//			dateStr_hour = tuple.getStringByField("dateStr_hour");
//			sourceAreaId = tuple.getStringByField("sourceAreaId");
//			areaName = tuple.getStringByField("areaName");
//			sourceAreaName = tuple.getStringByField("sourceAreaName");
//			sourceCountry = tuple.getStringByField("sourcecountry");
//			dnStreamOctets = tuple.getLongByField("dnstreamoctets");
//			dnStreamPacket = tuple.getLongByField("dnstreampacket");

//			pvKey = sip + "_" +dateStr_d;
//			int pv = triplePV;
//			if(pvMap.get(pvKey) != null) {
//				pv = pvMap.get(pvKey) + pv;
//			}
			
//			String debugString = "MetisStat countLocal <<<<<<<<< " + dip + " " + sip + " " + abnormaltype + " " + upStreamOctets + " " + upStreamPacket;
//			Logger.getRootLogger().info(debugString);
			
			sessKey = dateStr_d + "_" + sip + "_" + abnormaltype;
			sessReqKey = sip + "_" + dateStr_d + "_" + dip + "_" + abnormaltype;
			if(upStreamOctetsMap.get(sessKey) != null) {
				upStreamOctetsMap.put(sessKey, upStreamOctetsMap.get(sessKey) + upStreamOctets);
			} else {
				upStreamOctetsMap.put(sessKey,upStreamOctets);
			}
//			if(upStreamPacketMap.get(sessKey) != null) {
//				upStreamPacketMap.put(sessKey, upStreamPacketMap.get(sessKey) + upStreamPacket);
//			} else {
//				upStreamPacketMap.put(sessKey, upStreamPacket);
//			}
			
			if(sessReqOctetsMap.get(sessReqKey) != null) {
				sessReqOctetsMap.put(sessReqKey, sessReqOctetsMap.get(sessReqKey) + upStreamOctets);
			} else {
				sessReqOctetsMap.put(sessReqKey,upStreamOctets);
			}
//			if(upStreamPacketMap.get(sessReqKey) != null) {
//				upStreamPacketMap.put(sessReqKey, upStreamPacketMap.get(sessReqKey) + upStreamPacket);
//			} else {
//				upStreamPacketMap.put(sessReqKey, upStreamPacket);
//			}
//			if(dnStreamOctetsMap.get(sessKey) != null) {
//				dnStreamOctetsMap.put(sessKey, dnStreamOctetsMap.get(sessKey) + dnStreamOctets);
//			}
//			if(dnStreamPacketMap.get(sessKey) != null) {
//				dnStreamPacketMap.put(sessKey, dnStreamPacketMap.get(sessKey) + dnStreamPacket);
//			}
//			pvMap.put(pvKey, pv);
			
		}
	}

	@Override
	public void cleanup() {
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
